package cn.zyq.charging.order.mqtt;
import cn.zyq.charging.common.pojo.JsonResult;
import cn.zyq.charging.common.pojo.param.ChargingProcessParam;
import cn.zyq.charging.common.pojo.utils.JsonUtils;
import cn.zyq.charging.common.pojo.vo.ChargingProcessVO;
import cn.zyq.charging.order.constant.Constant;
import cn.zyq.charging.order.constant.MqttConstant;
import cn.zyq.charging.order.constant.OrderStatusConstant;
import cn.zyq.charging.order.mqtt.MqttContext;
import cn.zyq.charging.order.pojo.param.DeviceStartResultParam;
import cn.zyq.charging.order.pojo.po.ChargingBillFailPO;
import cn.zyq.charging.order.pojo.po.ChargingBillSuccessPO;
import cn.zyq.charging.order.pojo.po.ChargingProcessPO;
import cn.zyq.charging.order.service.OrderService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.BeanUtils;

import java.math.BigDecimal;

//MqttConsumer类会在本模块的MqttConfiguration中创建并纳入到spring容器
// 还会用"orderServiceClient"指定mqttClientId。 可以在
// http://localhost:18083/#/clients中看到此mqtt的客户端(同时能看到emqx.fx客户端软件客户端)

/**
 * 消息的消费者：
 * 用来消费设备发送的消息(接收设备发送的消息"开始充电" 指令执行成功， 或执行失败)
 *      设备使用充电枪发送的消息给订单模块(订单模块就是消费者，设备就是生产者)
 *      订单如果给充电桩发送的消息(那订单就是生产者， 充电桩就是消费者)
 * 有四个回调方法： connectComplete(重点关注)、connectionLost、messageArrived(重点关注)、deliveryComplete
 * 在MqConfiguration配置类中创建了本类MqttConsumer对象并设置给了mqttClient对象。
 */
@Slf4j //本类的方法是在MqttClient对象回调执行。
//@Component//MqttConsumer类需要自己在MqttConfiguration类的getMqttClient()方法中
//自己创建MqttConsumer类的对象，并创建MqttConsumer类对象设置为MqttConsumer对象的回调对象。
// (所以不需要标注@Component<此注解是让spring容器创建对象>， 所以MqttClient就不能注入对象<因为MqttConsumer是自己new对象的>)
public class MqttConsumer  implements MqttCallbackExtended {//MQTT: 是消息队列传输协议

    //@Autowired//因为要在MqttConfig配置类中new MqttConsumer的对象(没有让容器new对象，所以无法注入依赖对象)，所以无法注入(就改成下边构造方法注入的方式)
    private MqttClient mqttClient;

    public MqttConsumer(){}

    public MqttConsumer(MqttClient mqttClient){
        this.mqttClient=mqttClient;
    }

    private MqttContext mqttContext;

    //为了避免MqttConsumer的构造方法参数过多(有时还会给此类的构造方法不断添加参数)，
    //就使用MqttContext封装多个有可能被MqttConsumer类使用的对象(比如orderSuccessMapper)。
    public MqttConsumer(MqttClient mqttClient,MqttContext mqttContext){
        this.mqttClient=mqttClient;
        this.mqttContext=mqttContext;
    }

    public void setMqttClient(MqttClient mqttClient) {
        this.mqttClient = mqttClient;
    }



    /**连接完成，表示已经和EMQX消息中间件连接成功*/
    public void connectComplete(boolean reconnect, String serverURI) {//reconnect是是否重连的意思
        //连接成功后，接收消息(接收消息和发送消息的主题应该不一样，否则就区分不了是发送消息还是接收消息了)
        try {
            //给消息队列的客户端设置一个主题
            //1.和EMQX连接成功后，订阅"/topic/start/result"主题 ，来消费"开始充电指令"是否成功的消息(接收消息)
            log.debug("MQTTConsumer连接MQTTServer成功,订阅topic:{}", MqttConstant.TOPIC_CHARGING_RESULT);
            //mqttClient.subscribe("/topic/start/result", 0);
            mqttClient.subscribe(MqttConstant.TOPIC_CHARGING_RESULT);
            log.debug("MQTTConsumer连接MQTTServer成功,订阅topic:{}成功", MqttConstant.TOPIC_CHARGING_RESULT);

            //MqttConsumer.java的 方法中，订阅"/topic/charging/process"成功后，可以用Mqtt.fx客户端软件，发送  {"gunId":1} 消息给Emqx服务器。cx gmfndghjklk有覅回复申请书
            //2.订阅开始充电指令是否成功的消息后，再订阅"/topic/charging/process"主题 ，来消费"充电进度"的消息(接收消息)
            log.debug("MQTTConsumer连接MQTTServer成功,订阅topic:{}", MqttConstant.TOPIC_CHARGING_PROCESS);
            //mqttClient.subscribe("/topic/start/result", 0);
            mqttClient.subscribe(MqttConstant.TOPIC_CHARGING_PROCESS);
            log.debug("MQTTConsumer连接MQTTServer成功,订阅topic:{}成功", MqttConstant.TOPIC_CHARGING_PROCESS);
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }


    /**连接断开执行的方法*/
    public void connectionLost(Throwable throwable) {
        log.debug("连接断开--MQTT Connection lost");
    }


    private OrderService orderService;




    /*public MqttConsumer(MqttClient mqttClient,OrderService orderService){
        this.mqttClient=mqttClient;
        this.orderService=orderService;
    }*/

    /**接收到消息后执行此方法 ,MqttMessage是一个json格式的对象*/
    public void messageArrived(String topic, MqttMessage mqttMessage) {
        log.debug("MqttConsumer: 消息到达：topic-{},message-{}",topic,mqttMessage);
        //1.如果接收到的消息是"开始充电指令是否成功"，则调用handleChargingResult方法处理开始充电是否成功的结果
        if(MqttConstant.TOPIC_CHARGING_RESULT.equals(topic)){
            log.debug("处理开始充电结果消息:{}",mqttMessage);
            handleStartChargingResult(mqttMessage);//处理开始充电是否成功的结果
            log.debug("处理开始充电结果消息:{}，成功 √",mqttMessage);
            //2.如果接收到的消息是"充电进度"的消息，则调用handleChargingProcess方法处理开始充电是否成功的结果
        }else if(MqttConstant.TOPIC_CHARGING_PROCESS.equals(topic)){
            log.debug("处理开始充电进度消息:{}",mqttMessage);
            handleChargingProcess(mqttMessage);//处理正在充电的场景及充电完成的场景方法。
            log.debug("处理开始充电进度消息:{}，成功 √",mqttMessage);
        }
    }



    /**handleChargingResult方法： 处理开始充电是否成功的结果*/
    private void handleStartChargingResult(MqttMessage mqttMessage) {
        DeviceStartResultParam deviceStartResultParam =null;//讲师用的类是ChargingResultParam(我的代码换了一个类名)
        try {
            log.debug("开始处理充电结果消息:{}",mqttMessage.toString());
            deviceStartResultParam = JsonUtils.fromJsonToObject(mqttMessage.toString(), DeviceStartResultParam.class);
            log.debug("转化得到的deviceStartResultParam:{}", deviceStartResultParam);
        }catch (Exception e){
            log.warn("设备发送的开始充电结果消息格式有问题，请检查!", mqttMessage);
        }
        log.debug("消息到达",deviceStartResultParam);
        //判断接收到的对象是否存在，如果存在就处理
        if(deviceStartResultParam !=null){
            String result= deviceStartResultParam.getResult();//获取对象中的响应结果字符串(返回结果)
            //当Mqtt的客户端软件
            if(Constant.RESULT_START_CHARGING_SUCCESS.equals(result)){
                //如果返回的结果是发送充电指令成功， 则就调用orderService的saveSuccessOrder()将成功订单信息保存。
                log.debug("充电桩开始充电{}", deviceStartResultParam);

                ChargingBillSuccessPO chargingBillSuccessPO = new ChargingBillSuccessPO();
                chargingBillSuccessPO.setBillId(deviceStartResultParam.getOrderNo());
                //ORDER_STATUS_PROGRESS: 充电中//表示充电成功生成的订单状态值是1  (1表示充电中)
                chargingBillSuccessPO.setBillStatus(OrderStatusConstant.ORDER_STATUS_PROGRESS);
                //chargingBillSuccessPO的插入时间和修改时间默认是当前的时间值(DEFAULT CURRENT_TIMESTAMP)
                //为了避免本类MqttConsumer的构造方法参数过多(有时还会给此类的构造方法不断添加参数)，
                // 就使用MqttContext封装多个有可能被MqttConsumer类使用的对象(比如orderSuccessMapper)。
                log.debug("保存订单数据:{}",chargingBillSuccessPO);
                //mqttContext.getOrderSuccessMapper().insert(chargingBillSuccessPO);
                try {
                    //如果Emqx客户端重复返回 开始充电成功的反馈消息， 则会导致mysql重复保存同一个订单数据。
                    //解决办法1： 给订单表charging_bill_success或者charging_bill_fail表的订单编号列添加唯一索引(保证订单数据不重复)。
                    //此时如果保存重复的订单数据mysql就会通过mysql的驱动让mybatis抛出异常。
                    //这个异常我们可以处理(订单服务出现异常时调用设备服务给通知后台维修人员检查设备)
                    //这个异常也可以不用代码处理，而用普罗米修斯这种监控工具来监控我们的服务，如果出现多次同样的异常就通知维修人员。
                    //解决办法2： 我们也可以在保存一个订单之前先查询订单表是否有数据，如果没有数据就插入一条数据，否则就不插入数据。
                    //但是这个方式有一个原子性的问题不好处理(课堂上没讲这个处理方式，因为比较复杂<在高并发的情况下很难保证不重复>)
                    //解决办法3：redis布隆过滤器记录每次保存的状态， 如果已经在布隆过滤器中已经存在这个保存订单的状态值了，就不再执行了。
                    //布隆过滤器就类似于一个map集合，可以将订单id作为1，将规定的一个值作为value(比如1)。 第二次创建这个订单就可以去布隆过滤器中查询是否处理过这个订单
                    //处理过就不再处理这一条订单了
                    mqttContext.getOrderRepository().saveSuccessOrder(chargingBillSuccessPO);//OrderRepositoryImpl中调用MyBatisPlus的方法保存数据
                }catch(Exception e){
                    log.error("保存订单异常",e);
                }
                log.debug("保存订单数据:{}，成功",chargingBillSuccessPO);
                //创建订单成功记录
                //orderService.saveSuccessOrder();
            }else {//如果订单服务接收不到MQTT物联网中间件(模拟充电桩)返回的消息(执行充电指令成功或者失败)， 则说明设备故障。
                log.debug("开始充电后的反馈状态是： '开始充电失败'(充电桩故障)");
                ChargingBillFailPO chargingBillFailPO = new ChargingBillFailPO();
                chargingBillFailPO.setBillId(deviceStartResultParam.getOrderNo());
                chargingBillFailPO.setFailDesc("设备自检失败, 设备有响应，但是充不了电");
                //TODO: 后续处理无响应的设备

                //如果返回的结果不是发送充电指令成功， 则就调用orderService的saveFailOrder()将失败订单信息保存。
                log.debug("充电桩开始充电失败(充电桩故障<启动充电失败>)，保存失败订单数据：{}" , chargingBillFailPO);
                mqttContext.getOrderRepository().saveFailOrder(chargingBillFailPO);//用mqttContext的OrderFailMapper的insert方法保存失败的订单数据
                log.debug("保存失败订单数据：{}成功", chargingBillFailPO);
                mqttContext.getOrderRepository().saveFailOrder(chargingBillFailPO);//OrderRepositoryImpl中调用MyBatisPlus的方法保存数据
            }
        }
    }


    /**
     * 设备给订单服务同步充电进度
     * 处理: 1.设备给订单服务同步充电进度信息。
     *      2.当充电完处理成计算充电费用。
     * @param mqttMessage
     */
    private void handleChargingProcess(MqttMessage mqttMessage) {
        //1.读取接收的json消息中包含的充电进度数据
        ChargingProcessParam chargingProcessParam =null;
        try {//从Mqtt返回的充电进度消息对象MqttMessage中获取充电进度对象chargingProcessParam
            log.debug("接收到设备发送的充电进度消息:{}",mqttMessage.toString());
            chargingProcessParam = JsonUtils.fromJsonToObject(mqttMessage.toString(), ChargingProcessParam.class);
            log.debug("消息转换为ChargingProcessParam:{}",chargingProcessParam);
        }catch (Exception e){
            //e.printStackTrace();
            log.warn("设备发送的充电进度消息格式有问题，请检查!", mqttMessage);
        }
        log.debug("通过open-feign结合costClient调用计价服务CostService 计算价格 .入参:{}",chargingProcessParam);
        //2.调用计价服务计算电费(在getOrderCost()中通过open-feign的costClient 调用计价服务CostService 计算价格
        //<获取ChargingProcessVO对象<封装了充电费用信息>>)
        ChargingProcessVO orderCostChargingProcessVO=getOrderCost(chargingProcessParam);
        log.debug("通过open-feign结合costClient调用计价服务CostService 计算价格 成功.入参:{},出参:{}",chargingProcessParam,orderCostChargingProcessVO);
        if(orderCostChargingProcessVO!=null){
            //获取充电进度对象orderCostChargingProcessVO中的充电总金额数据
            BigDecimal totalCost=orderCostChargingProcessVO.getTotalCost();
            //打印计算的总费用。
            log.debug("当前花费总金额:{}",totalCost);
            //3. 保存充电进度数据
            saveChargingProcess(chargingProcessParam);
            //4 推送充电数据给用户 推送(WebSocket技术实现)
            syncChargingProcessToUser(orderCostChargingProcessVO);
            //5 充电过程中如果钱不够了,如何处理?
            //  5.1 方式一： 给用户发通知,钱不够了,就结束充电，引导用户充值 (不人性化)
            //  5.2 方式二： 钱不够 继续充 直到充满为止 记录下用户欠款
            assert chargingProcessParam != null;
            checkUserBalance(chargingProcessParam.getUserId(), totalCost);
            //6 判断温度
            //  6.1 大于 40度 小于 50度 B级告警  推送给用户 用户可以选择是否继续充, 还是结束充电。
            //  6.2 大于 50度  A级告警  推送给用户 停止充电
            checkDeviceStatus(chargingProcessParam.getTemperature());
            //7 判断是否充满
            //   6.1 停止充电 更新订单状态为 订单结束
            checkChargingStatus(chargingProcessParam.getIsFull());//判断是否充满
        }else{//讲师代码里没写这个else{}分支
            //讲师代码里没写这个else{}分支
            //订单服务需要对计价服务返回的结果进行判断 如果 为 null 表示计算价格失败 订单需要通知设备停止充电
            log.debug("订单服务需要对计价服务返回的结果进行判断 如果 为 null 表示计算价格失败 订单需要通知设备停止充电");
        }
    }


    /**
     * 通过open-feign的costClient 调用计价服务  计算充电总费用
     * @param chargingProcessParam
     * @return
     */
    private ChargingProcessVO getOrderCost(ChargingProcessParam chargingProcessParam){
        JsonResult<ChargingProcessVO> result=mqttContext.getCostClient().calculateCost(chargingProcessParam);
        //CostClient是通过OpenFeign调用计费服务的计费服务接口CostClient
        if(result!=null){
            ChargingProcessVO data=result.getData();
            if(data !=null){
                return data;
            }
        }
        return null;
    }



    /**
     * 保存充电进度(如果保存到mysql中，则数据量太庞大，并发量太大了)
     * 存储 存到mysql 是否可以
     *   答案：不太行mysql性能较差  量太大 100 * 30s 一次 1分钟 2百次  1分钟产生2百条数据  平均充满电需要 2个小时  2W4(2小时产生2万4千条)  * 12(1天)  == 288000(一天产生288万条数据)
     * 结论：将同步的充电数据保存到ES分布式中间件中。 ES是一个分布式搜索引擎
     * 注意： 数据库每秒也就是几百条并发量，不会上千。
     *
     * 时序数据：基本只会频繁的写入(不会更新)，少量读取(通常出现在电力行业、 化工行业、气象行业、 地理信息等各类型实时监测检查分析设备所采集)。
     * 时间 数据
     *   10:01 充电数据
     *   10:02 充电数据
     *   10:03 充电数据
     *
     * 天气数据也是时序数据。
     *   10:01  风力 5
     *   10:02  6   更新不会
     *   10:03  10
     *
     * 特点: 按照时间 序列  数据量大
     * 读写特点: 高频率的写 相对写操作来说 读操作比较少, 而且不会更新数据。
     */
    private void saveChargingProcess(ChargingProcessParam chargingProcessParam) {
        log.debug("保存充电进度数据到ES:{}",chargingProcessParam);
        //创建ChargingProcessPO对象，用于封装同步的充电数据对象ChargingProcessParam的信息。
        ChargingProcessPO chargingProcessPO = new ChargingProcessPO();
        chargingProcessPO.setId(System.currentTimeMillis());//设置时间
        BeanUtils.copyProperties(chargingProcessParam,chargingProcessPO);//数据拷贝
        //从mqttContext获取ChargingProcessRepository对象 并调用其save()方法将chargingProcessPO保存到ES中。
        try {
            mqttContext.getChargingProcessRepository().save(chargingProcessPO);//保存chargingProcessPO数据到ES
        }catch (Exception e){
            e.printStackTrace();
        }
        log.debug("保存充电进度数据成功{}", chargingProcessParam);
    }



    /**
     *  同步充电数据给用户(充电总费用、总度数、温度等信息)
     * @param orderCostChargingProcessVO 同步的充电数据对象
     */
    private void syncChargingProcessToUser(ChargingProcessVO orderCostChargingProcessVO){
        log.debug("同步充电数据到用户手机端:{}",orderCostChargingProcessVO);
        String json=JsonUtils.toJson(orderCostChargingProcessVO);//将orderCostChargingProcessVO对象转换为json字符串
        //使用webSocket 将json字符串推送给用户手机端。
        log.debug("同步充电数据到用户手机端22:{}",json);
        mqttContext.getWebSocketServer().sendMessage(orderCostChargingProcessVO.getUserId(),json);
        //这个sendMessage()推送消息的功能 后续还可以集成第三方的消息推送服务比如 极光、网易云信等推送服务。
        //TODO: 后续还可以将上一行用WebSocket发送消息到Rabbitmq， 然后用客户端订阅此消息来实现异步消息推送。
        //区别1：这种模式下服务端不用在意客户端是否在线，只是将消息发送到RabbitMQ即可。
        //区别2：增加了RabbitMQ，可以提供消息持久化、重试、死信队列等机制，确保消息不丢失(更加可靠)。
        //区别3：多个 WebSocket 服务器可以共享 RabbitMQ 队列，避免单点压力。
        //区别4：但是加了RabbitMQ消息延迟性更大(因为结构变的更复杂了)。 这种模式适用于: 分布式系统，需要多个 WebSocket 服务器协同工作(百万级并发量<不加则是10万级并发量>)。
        //因为上一行代码是同步执行的，消息没发送成功的话不会执行后续操作。
        //WebSocket还可以结合netty来实现高性能的网络通信功能。  netty是一个高性能的网络通信容器(类似于tomcat),
        //可以提供高性能的网络通信处理服务。(netty比tomcat并发高、传输快)
        //https://blog.csdn.net/m0_63251896/article/details/134518001?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522552b53f484b749dabda346994eea20b0%2522%252C%2522scm%2522%253A%252220140713.130102334..%2522%257D&request_id=552b53f484b749dabda346994eea20b0&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~blog~baidu_landing_v2~default-1-134518001-null-null.nonecase&utm_term=netty%E6%95%B4%E5%90%88websocket%28%E5%AE%8C%E7%BE%8E%E6%95%99%E7%A8%8B%29&spm=1018.2226.3001.4450
    }


    //注意：测试余额不足的方式是
    //1.启动ES、 Redis、 mysql、nacos、emqx、订单服务、计费服务、用户服务、设备服务、gateWay模块
    //2.启动mqtt.fx客户端工具并连接物联网emqx消息中间件服务器。 并发送topic为"/topic/charging/process"
    //  的消息给订单服务(具体消息内容如下<充电度数比较大，总花费大于余额>)， 发送的消息中的数据计算出的充电总费用>余额，就会出现余额不足的情况。
    //{
    //  "orderNo":"10000_54_1746585230507",
    //  "pileId":1 ,
    //  "gunId":1,
    //  "stationId" :5,
    //  "chargingCapacity":10000,
    //  "temperature":20,
    //  "isFull":false
    //}
    /**
     * 检查用户余额是否够支付目前的充电总金额。//用户余额需要余额需要通过用户服务来获取
     * @param userId 用户编号
     * @param totalCost 充电总金额(花费的总金额)
     */
    private void checkUserBalance(Integer userId,BigDecimal totalCost){
        System.err.println("检查用户余额");
        //通过OpenFeign调用用户服务获取用户余额的方法getUserBalance()获取用户余额信息
        JsonResult<BigDecimal> userBalanceResult= mqttContext.getUserClient().getUserBalance(userId);
        if(userBalanceResult!=null){
            BigDecimal userBalance = userBalanceResult.getData();
            //用户余额如果小于当前充电的总花费，则提示用户余额不足，请充值。
            //用户余额 - 充电总花费
            if(userBalance.subtract(totalCost).floatValue() <= 0 ){
                log.debug("给用户发送通知： 余额不足，即将停止充电，请及时充值。当前余额{}，总共消费金额{}",userBalance,totalCost);
            }
        }
    }

    /**
     * 检查设备状态： 目前只检查温度信息。   (后续可以继续再实现检查电压等信息)
     */
    private void checkDeviceStatus(float temperature){
        if(temperature >  MqttConstant.A_WARNING){//大于50度
            log.debug("温度超过:{},给用户发送A类告警,直接停止充电,实际温度:{}",MqttConstant.A_WARNING,temperature);
            //通知用户 推送消息给用户  mqttContext.getWebSocketServer().sendMessage
        }else if(temperature > MqttConstant.B_WARNING){//大于40度
            log.debug("温度超过:{},给用户发送B类告警,设备不停止充电,让用户选择是否继续充电,实际温度:{}",MqttConstant.B_WARNING,temperature);
            //通知用户 推送消息给用户  mqttContext.getWebSocketServer().sendMessage
        }
    }


    /**
     * 检查设备状态： 目前只检查温度信息。   (后续可以继续再实现检查电压等信息)
     */
    private void checkChargingStatus(Boolean isFull){
        if (isFull) {
            log.debug("已经充满,更新订单状态,更新余额");
            //给用户发送充电结束 和 费用信息
            // mqttContext.getWebSocketServer().sendMessage
        }
    }




    /**消息接收完毕后执行的方法*/
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        log.debug("MQttConsumer--消息接受完毕.");
    }

}
